package srzy.socket

import srzy.common.Logging
import srzy.common.Config
import java.net._
import srzy.rb._
import srzy.rb.RBMsg
import java.util.concurrent.locks.ReentrantLock
import java.io.IOException

class SocketConn(serverIp: String, port: Int) extends srzy.common.Logging {

  val socket = connect
  val outStream = socket.getOutputStream()
  val inStream = socket.getInputStream()
  val readLock = new ReentrantLock()
  val writeLock = new ReentrantLock()

  def connect: Socket = {
    try {
      val s = new Socket()
      s.setSoLinger(true, 2)
      s.setSoTimeout(5000)
      s.setSendBufferSize(32 * 1024)
      s.setReceiveBufferSize(32 * 1024)
      s.setTcpNoDelay(true)
      s.connect(new InetSocketAddress(serverIp, port))
      logger.info("Established connection to " + serverIp + ":" + port
        + " connect success!")
      s
    } catch {
      case e: Exception =>
        logger.error("Established connection to " + serverIp + " : " + port
          + " connect failure!" + e.getMessage(), e)
        throw e
    }
  }

  def authentication = {
    write(SocketConn.authenticationMsg)
    logger.debug("AuthenticationReq write over!")
    val msg = read()
    val errCode = Integer.parseInt(getErrorCode(msg))

    if (errCode != 0) {
      throw new Exception("Authentication failure : " + getSessionId(msg))
    }
    logger.info("Authentication success")
    logger.info("Authentication msg : " + msg)
  }

  def replyHeartbeat = {
    write(SocketConn.heartMsg)
    logger.debug("Reply Heartbeat")
  }

  def write(msg: String) = {
    val b = RBMsg.pack(msg)
    writeLock.lock()
    try {
      outStream.write(b, 0, b.length)
      outStream.flush()
    } finally {
      writeLock.unlock()
    }
  }

  def read(): String = {
    readLock.lock()
    try {
      val header = this.read(RBMsg.HEAD_SIZE)
      val body = this.read(RBMsg.getMsgBodyLen(header))
      return RBMsg.getMsgBody(body)
    } finally {
      readLock.unlock()
    }
  }

  def read(len: Int): Array[Byte] = {
    //    logger.debug("read in")
    val obj = new Array[Byte](len)
    var readed = 0
    var reading = 0
    var left = obj.length
    while (left > 0) {
      //      logger.debug("readed:" + readed + ", reading:" + reading + ", left:" + left)
      if (Thread.currentThread().isInterrupted()) {
        logger.debug("Thread is interrupted!")
        throw new InterruptedException()
      }
      //      try {
      reading = inStream.read(obj, readed, left)
      if (reading == -1) {
        throw new IOException("read failed reading == -1")
      }
      readed += reading
      left -= reading
      //      } catch {
      //        case ste: SocketTimeoutException =>
      //        //          logger.debug("Socket read timeout")
      //      }
    }
    logger.debug("readed:" + readed + ", reading:" + reading + ", left:"
      + left)
    //    logger.debug("read done")
    return obj
  }

  def close() = {
    logger.info("Close SocketConn start!")
    try {
      socket.close()
      outStream.close()
      inStream.close()
    } catch {
      case e: Exception => logger.error("Close the connection failed : " + e.getMessage())
    }
    logger.info("Close SocketConn over!")
  }
}

object SocketConn extends srzy.common.Logging {

  val heartMsg = Config.getValue("HeartMsgResp", "BodyMsg")
  val authenticationMsg = Config.getValue("AuthenticationMsg", "BodyMsg")

  val serverIp = Config.getValue("root", "config", "SERVER_IP")
  val port = Integer.valueOf(Config.getValue("root", "config", "SERVER_PORT"))
  //val SocketTimeOutTime = Integer.valueOf(config.getValue(channelName,
  // "TIMEOUTTIME"))

  def apply(): SocketConn = {
    val s = new SocketConn(serverIp, port)
    s.authentication
    s
  }

}